package zookeeper.impl;

import lombok.Data;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

// zookeeper 可重入锁实现
public class ZKReentrantLock implements Lock {

    private String defaultLockPath = "/dorm_locks";
    private String zkServer = "127.0.0.1:2181";

    private ZkClient zkClient;

    ConcurrentHashMap<Thread, TData> threadMap = new ConcurrentHashMap<>();

    class TData {
        //当前锁节点
        private String currentPath;

        //线程获取锁的次数
        private AtomicInteger lockCount = new AtomicInteger(0);

        public String getCurrentPath() {
            return currentPath;
        }

        public void setCurrentPath(String currentPath) {
            this.currentPath = currentPath;
        }

        public AtomicInteger getLockCount() {
            return lockCount;
        }

        public void setLockCount(AtomicInteger lockCount) {
            this.lockCount = lockCount;
        }
    }

    //当前获取锁的线程
    private static volatile Thread currentThread;

    public ZKReentrantLock() {
        super();
        this.zkClient = new ZkClient(zkServer);
        this.zkClient.setZkSerializer(new MyZkSerializer());

        //初始化一个默认的锁节点
        if (!zkClient.exists(defaultLockPath)) {
            zkClient.createPersistent(defaultLockPath);
        } else {
            zkClient.deleteRecursive(defaultLockPath);
            zkClient.createPersistent(defaultLockPath);
        }

    }

    /**
     * 如果获取锁失败，应该阻塞，直到获取锁为止
     */
    @Override
    public void lock() {
        if (!tryLock()) {
            //获取锁失败
            lock();
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock() {
        return tryAcquire(1);
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        tryRelease(1);
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    private boolean tryAcquire(int acquires) {

        final Thread thread = Thread.currentThread();
        System.out.println(thread.getName() + "tryAcquire");

        if (thread == getCurrentThread()) {
            // 当前线程已经获取锁 再次获取属于可重入
            int count = threadMap.get(thread).getLockCount().incrementAndGet();
            System.out.println(getCurrentThread().getName() + "获取锁成功，当前获取锁次数：" + count);
            return true;
        } else {
            //没有线程获取锁或者获取锁的不是当前线程
            if (compileAndSetLockCount(0, acquires, thread)) {
                //当前线程获取锁成功
                return true;
            }
        }

        //已有线程获取锁且不是当前线程
        return false;
    }

    private boolean tryRelease(int release) {

        Thread thread = Thread.currentThread();
        System.out.println(thread.getName() + "tryRelease");
        //不是当前线程获取锁
        if (getCurrentThread() != thread) {
            return false;
        }
        //剩余锁次数
        int count = threadMap.get(thread).getLockCount().decrementAndGet();
        if (count == 0) {
            //锁全部释放
            setCurrentThread(null);
            //删除当前节点
            zkClient.delete(threadMap.get(thread).getCurrentPath());
            threadMap.remove(thread);
            System.out.println("锁释放成功！" + thread.getName());
            return true;
        }
        System.out.println("锁释放成功！" + thread.getName() + "剩余次数：" + count);
        return false;
    }

    private boolean compileAndSetLockCount(int expect, int update, Thread thread) {

        System.out.println(thread.getName() + "compileAndSetLockCount");
        if (threadMap.get(thread) == null) {
            threadMap.put(thread, new TData());
            String currentPath = this.zkClient.createEphemeralSequential(defaultLockPath + "/", 1);
            threadMap.get(thread).setCurrentPath(currentPath);
        }
        //获取所有节点
        List<String> children = this.zkClient.getChildren(defaultLockPath);
        //排序
        Collections.sort(children);

        System.out.println(thread.getName() + "--currentPath:" + threadMap.get(thread).getCurrentPath());
        System.out.println(children);
        //判断当前节点是否为最小节点
        if (threadMap.get(thread).getCurrentPath().equals(defaultLockPath + "/" + children.get(0))) {
            System.out.println(thread.getName() + "--LockCount:" + threadMap.get(thread).getLockCount().get() + "--currentPath:" + threadMap.get(thread).getCurrentPath());
            int count = threadMap.get(thread).getLockCount().incrementAndGet();
            currentThread = thread;
            System.out.println(thread.getName() + "获取锁成功，当前获取锁次数：" + count);
            return true;
        }
        // /dorm_locks/xxx
        int curIndex = children.indexOf(threadMap.get(thread).getCurrentPath().substring(defaultLockPath.length() + 1));

        System.out.println(thread.getName() + "--curIndex:" + curIndex);
        //前一个节点
        String beforePath = defaultLockPath + "/" + children.get(curIndex - 1);
        waitForLock(beforePath);

        System.out.println(thread.getName() + "--beforePath:" + beforePath);
        return false;
    }

    private void waitForLock(String beforePath) {
        CountDownLatch latch = new CountDownLatch(1);
        IZkDataListener listener = new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                System.out.println("节点变更！");
            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
                System.out.println("节点被删除，开始抢锁！" + s);
                latch.countDown();
            }
        };

        //监听前一个节点的删除
        this.zkClient.subscribeDataChanges(beforePath, listener);

        //前一个节点存在，阻塞当前线程
        if (this.zkClient.exists(beforePath)) {
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        this.zkClient.unsubscribeDataChanges(beforePath, listener);
    }

    public static Thread getCurrentThread() {
        return currentThread;
    }

    public static void setCurrentThread(Thread currentThread) {
        ZKReentrantLock.currentThread = currentThread;
    }

    public static void main(String[] args) {

        ZKReentrantLock lock = new ZKReentrantLock();

        int num = 3;
        CyclicBarrier cyclicBarrier = new CyclicBarrier(num);
        CommonResources resources = new CommonResources();
        for (int i = 0; i < num; i++) {
            new Thread(new Runnable() {

                @Override
                public void run() {

                    System.out.println(Thread.currentThread().getName() + "： 我准备好了");

                    try {
                        cyclicBarrier.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                    for (int i = 0; i < 30; i++) {
                        lock.lock();
                        resources.sale();
                        lock.unlock();
                        try {
                            TimeUnit.MILLISECONDS.sleep(new Random().nextInt(100));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
        }

    }
}
